package com.djf.flinkcdc.controller;

import com.djf.flinkcdc.service.CanalStyleSyncService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;

/**
 * Canal 风格数据同步控制器
 * 提供 Canal 风格的无感知数据同步管理接口
 * 
 * 核心功能说明：
 * 1. 事件监听：实时监听 MySQL binlog 变化，类似 Canal 的工作方式
 * 2. 无感同步：应用无需感知数据变更，自动捕获增删改操作
 * 3. 事件分发：支持多个监听器同时处理数据变更事件
 * 4. 数据缓存：缓存表数据状态，支持变更前后数据对比
 * 5. 统计监控：提供详细的操作统计和监控信息
 */
@Slf4j
@RestController
@RequestMapping("/api/canal")
public class CanalController {

    @Autowired
    private CanalStyleSyncService canalSyncService;

    /**
     * Canal 服务健康检查
     * GET /api/canal/health
     */
    @GetMapping("/health")
    public ResponseEntity<Map<String, Object>> health() {
        Map<String, Object> response = new HashMap<>();
        response.put("service", "Canal Style Sync Service");
        response.put("status", "healthy");
        response.put("running", canalSyncService.isCanalSyncRunning());
        response.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        response.put("description", "Canal 风格无感知数据同步服务");
        
        log.info("🏥 Canal 服务健康检查 - 状态: {}", canalSyncService.isCanalSyncRunning() ? "运行中" : "已停止");
        return ResponseEntity.ok(response);
    }

    /**
     * 获取 Canal 服务统计信息
     * GET /api/canal/statistics
     */
    @GetMapping("/statistics")
    public ResponseEntity<Map<String, Object>> getStatistics() {
        Map<String, Object> stats = canalSyncService.getCanalStatistics();
        stats.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        stats.put("description", "Canal 风格数据同步统计信息");
        
        log.info("📊 获取 Canal 统计信息 - 总事件: {}, 新增: {}, 更新: {}, 删除: {}", 
                stats.get("totalEvents"), stats.get("insertCount"), 
                stats.get("updateCount"), stats.get("deleteCount"));
        
        return ResponseEntity.ok(stats);
    }

    /**
     * 启动 Canal 风格数据同步
     * POST /api/canal/start
     */
    @PostMapping("/start")
    public ResponseEntity<Map<String, Object>> startCanalSync() {
        Map<String, Object> response = new HashMap<>();
        
        try {
            if (canalSyncService.isCanalSyncRunning()) {
                response.put("success", false);
                response.put("message", "Canal 风格同步服务已在运行中");
                response.put("status", "already_running");
                log.warn("⚠️ Canal 同步启动失败 - 服务已在运行中");
                return ResponseEntity.badRequest().body(response);
            }
            
            canalSyncService.startCanalSync();
            
            response.put("success", true);
            response.put("message", "Canal 风格数据同步启动成功");
            response.put("status", "started");
            response.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            response.put("features", getCanalFeatures());
            
            log.info("🚀 Canal 风格数据同步启动成功");
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            response.put("success", false);
            response.put("message", "Canal 风格数据同步启动失败: " + e.getMessage());
            response.put("status", "error");
            log.error("❌ Canal 同步启动失败: {}", e.getMessage(), e);
            return ResponseEntity.internalServerError().body(response);
        }
    }

    /**
     * 停止 Canal 风格数据同步
     * POST /api/canal/stop
     */
    @PostMapping("/stop")
    public ResponseEntity<Map<String, Object>> stopCanalSync() {
        Map<String, Object> response = new HashMap<>();
        
        try {
            if (!canalSyncService.isCanalSyncRunning()) {
                response.put("success", false);
                response.put("message", "Canal 风格同步服务未在运行");
                response.put("status", "not_running");
                log.warn("⚠️ Canal 同步停止失败 - 服务未在运行");
                return ResponseEntity.badRequest().body(response);
            }
            
            // 获取停止前的统计信息
            Map<String, Object> finalStats = canalSyncService.getCanalStatistics();
            
            canalSyncService.stopCanalSync();
            
            response.put("success", true);
            response.put("message", "Canal 风格数据同步停止成功");
            response.put("status", "stopped");
            response.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            response.put("finalStatistics", finalStats);
            
            log.info("🛑 Canal 风格数据同步停止成功 - 最终统计: {}", finalStats);
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            response.put("success", false);
            response.put("message", "Canal 风格数据同步停止失败: " + e.getMessage());
            response.put("status", "error");
            log.error("❌ Canal 同步停止失败: {}", e.getMessage(), e);
            return ResponseEntity.internalServerError().body(response);
        }
    }

    /**
     * 重启 Canal 风格数据同步
     * POST /api/canal/restart
     */
    @PostMapping("/restart")
    public ResponseEntity<Map<String, Object>> restartCanalSync() {
        Map<String, Object> response = new HashMap<>();
        
        try {
            log.info("🔄 开始重启 Canal 风格数据同步服务...");
            
            // 如果正在运行，先停止
            if (canalSyncService.isCanalSyncRunning()) {
                canalSyncService.stopCanalSync();
                Thread.sleep(2000); // 等待2秒确保完全停止
            }
            
            // 重新启动
            canalSyncService.startCanalSync();
            
            response.put("success", true);
            response.put("message", "Canal 风格数据同步重启成功");
            response.put("status", "restarted");
            response.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            response.put("features", getCanalFeatureslist());
            
            log.info("✅ Canal 风格数据同步重启成功");
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            response.put("success", false);
            response.put("message", "Canal 风格数据同步重启失败: " + e.getMessage());
            response.put("status", "error");
            log.error("❌ Canal 同步重启失败: {}", e.getMessage(), e);
            return ResponseEntity.internalServerError().body(response);
        }
    }

    /**
     * 获取 Canal 风格同步状态
     * GET /api/canal/status
     */
    @GetMapping("/status")
    public ResponseEntity<Map<String, Object>> getCanalStatus() {
        Map<String, Object> response = new HashMap<>();
        Map<String, Object> stats = canalSyncService.getCanalStatistics();
        
        response.put("running", canalSyncService.isCanalSyncRunning());
        response.put("statistics", stats);
        response.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        response.put("description", "Canal 风格数据同步状态信息");
        
        // 添加状态描述
        if (canalSyncService.isCanalSyncRunning()) {
            response.put("statusDescription", "🟢 Canal 风格同步服务正在运行，实时监听数据变更");
        } else {
            response.put("statusDescription", "🔴 Canal 风格同步服务已停止");
        }
        
        log.info("📋 获取 Canal 状态 - 运行状态: {}, 总事件: {}", 
                canalSyncService.isCanalSyncRunning(), stats.get("totalEvents"));
        
        return ResponseEntity.ok(response);
    }

    /**
     * 获取 Canal 风格功能说明
     * GET /api/canal/features
     */
    @GetMapping("/features")
    public ResponseEntity<Map<String, Object>> getCanalFeatures() {
        Map<String, Object> response = new HashMap<>();
        response.put("service", "Canal Style Data Sync Service");
        response.put("version", "1.0.0");
        response.put("features", getCanalFeatureslist());
        response.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        
        log.info("📖 获取 Canal 功能说明");
        return ResponseEntity.ok(response);
    }

    /**
     * 获取 Canal 风格核心功能列表
     */
    private Map<String, Object> getCanalFeatureslist() {
        Map<String, Object> features = new HashMap<>();
        
        // 事件监听功能
        Map<String, Object> eventListening = new HashMap<>();
        eventListening.put("name", "事件监听");
        eventListening.put("description", "实时监听 MySQL binlog 变化，类似 Canal 的工作方式");
        eventListening.put("capabilities", new String[]{
            "实时 binlog 监听", "自动事件捕获", "多表同时监听", "事件顺序保证"
        });
        features.put("eventListening", eventListening);
        
        // 无感数据同步
        Map<String, Object> transparentSync = new HashMap<>();
        transparentSync.put("name", "无感数据同步");
        transparentSync.put("description", "应用无需感知数据变更，自动捕获增删改操作");
        transparentSync.put("operations", new String[]{
            "INSERT - 新增操作监听", "UPDATE - 更新操作监听", "DELETE - 删除操作监听"
        });
        transparentSync.put("benefits", new String[]{
            "零代码侵入", "实时数据捕获", "完整变更记录", "数据一致性保证"
        });
        features.put("transparentSync", transparentSync);
        
        // 事件分发机制
        Map<String, Object> eventDispatch = new HashMap<>();
        eventDispatch.put("name", "事件分发机制");
        eventDispatch.put("description", "支持多个监听器同时处理数据变更事件");
        eventDispatch.put("features", new String[]{
            "多监听器支持", "异步事件处理", "错误隔离", "自定义监听器"
        });
        features.put("eventDispatch", eventDispatch);
        
        // 数据缓存与对比
        Map<String, Object> dataCache = new HashMap<>();
        dataCache.put("name", "数据缓存与对比");
        dataCache.put("description", "缓存表数据状态，支持变更前后数据对比");
        dataCache.put("capabilities", new String[]{
            "变更前数据缓存", "变更后数据记录", "数据差异对比", "状态追踪"
        });
        features.put("dataCache", dataCache);
        
        // 监控与统计
        Map<String, Object> monitoring = new HashMap<>();
        monitoring.put("name", "监控与统计");
        monitoring.put("description", "提供详细的操作统计和监控信息");
        monitoring.put("metrics", new String[]{
            "INSERT 操作计数", "UPDATE 操作计数", "DELETE 操作计数", 
            "总事件数统计", "监听器数量", "缓存表数量"
        });
        features.put("monitoring", monitoring);
        
        return features;
    }

    /**
     * 获取 Canal 风格使用示例
     * GET /api/canal/examples
     */
    @GetMapping("/examples")
    public ResponseEntity<Map<String, Object>> getCanalExamples() {
        Map<String, Object> response = new HashMap<>();
        
        // API 使用示例
        Map<String, Object> apiExamples = new HashMap<>();
        apiExamples.put("启动同步", "POST /api/canal/start");
        apiExamples.put("停止同步", "POST /api/canal/stop");
        apiExamples.put("重启同步", "POST /api/canal/restart");
        apiExamples.put("查看状态", "GET /api/canal/status");
        apiExamples.put("查看统计", "GET /api/canal/statistics");
        
        // 数据变更示例
        Map<String, Object> dataChangeExamples = new HashMap<>();
        dataChangeExamples.put("新增数据", "INSERT INTO users (name, email) VALUES ('张三', 'zhangsan@example.com')");
        dataChangeExamples.put("更新数据", "UPDATE users SET email = 'newemail@example.com' WHERE id = 1");
        dataChangeExamples.put("删除数据", "DELETE FROM users WHERE id = 1");
        
        // 监听器输出示例
        Map<String, Object> outputExamples = new HashMap<>();
        outputExamples.put("新增事件", "[2024-01-01 12:00:00] 🎯 Canal事件 | testdb.users | 新增 | 时间戳: 1704067200000");
        outputExamples.put("更新事件", "[2024-01-01 12:01:00] 🎯 Canal事件 | testdb.users | 更新 | 时间戳: 1704067260000");
        outputExamples.put("删除事件", "[2024-01-01 12:02:00] 🎯 Canal事件 | testdb.users | 删除 | 时间戳: 1704067320000");
        
        response.put("apiExamples", apiExamples);
        response.put("dataChangeExamples", dataChangeExamples);
        response.put("outputExamples", outputExamples);
        response.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        
        log.info("📚 获取 Canal 使用示例");
        return ResponseEntity.ok(response);
    }
}